package com.zlsy.springbootdemo.redis.consumer;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.time.Duration;

/**
 * @author zhouliang
 * @date 2020/12/15 17:24
 */
@Slf4j
@Component
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class StreamConsumerRunner implements ApplicationRunner, DisposableBean {

  private final RedisConnectionFactory redisConnectionFactory;
  private final ThreadPoolTaskExecutor poolTaskExecutor;
  private final StreamListener<String, MapRecord<String, String, String>> streamListener;
  //  private final RedisTemplate<String, Serializable> redisTemplate;

  private StreamMessageListenerContainer<String, MapRecord<String, String, String>>
      streamMessageListenerContainer;

  @Override
  public void destroy() throws Exception {
    try {
      this.streamMessageListenerContainer.stop();
    } catch (Exception e) {
      log.error("streamMessageListenerContainer destroy error:{}", e.getMessage());
    }
  }

  @Override
  public void run(ApplicationArguments args) throws Exception {
//    StreamMessageListenerContainer.StreamMessageListenerContainerOptions<
//            String, MapRecord<String, String, String>>
//        streamMessageListenerContainerOptions =
//            StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
//                .batchSize(10)
//                .executor(poolTaskExecutor)
//                .errorHandler(Throwable::printStackTrace)
//                .pollTimeout(Duration.ZERO)
//                .serializer(new StringRedisSerializer())
//                .build();
//    StreamMessageListenerContainer<String, MapRecord<String, String, String>>
//        streamMessageListenerContainer =
//            StreamMessageListenerContainer.create(
//                this.redisConnectionFactory, streamMessageListenerContainerOptions);
//    streamMessageListenerContainer.receiveAutoAck(
//        Consumer.from("group-1", "consumer-1"),
//        StreamOffset.create("myStreams", ReadOffset.latest()),
//        streamListener);
//    //    streamMessageListenerContainer.receive(
//    //        Consumer.from("group-1", "consumer-1"),
//    //        StreamOffset.create("myStreams", ReadOffset.lastConsumed()),
//    //        this.streamListener);
//
//    this.streamMessageListenerContainer = streamMessageListenerContainer;
//    this.streamMessageListenerContainer.start();
  }
}
